package com.ferry.registry.zk;

import com.ferry.common.util.IpUtil;
import com.ferry.registry.domain.registry.NodeEvent;
import com.ferry.registry.domain.registry.RegistryMetaData;
import com.ferry.registry.domain.registry.RegistryState;
import com.ferry.registry.domain.service.ServiceAddress;
import com.ferry.registry.domain.service.ServiceMetaData;
import com.ferry.registry.service.AbstractRegistryService;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import io.netty.util.internal.ConcurrentSet;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;

import static com.ferry.common.util.StackTraceUtil.stackTrace;

/**
 * Created by daiyong
 */
public class ZkRegistryServiceImpl extends AbstractRegistryService {

	Logger logger = LoggerFactory.getLogger(ZkRegistryServiceImpl.class);

	/**
	 * 当前机器ip地址
	 */
	private static final String ipAddress = IpUtil.getLocalAddress();

	/**
	 * zk会话时间
	 * 时间不能太长 否则临时节点保存时间太长会影响消费者调用
	 */
	private final int sessionTimeoutMs = 5 * 1000;
	/**
	 * zk链接时间
	 */
	private final int connectionTimeoutMs = 15 * 1000;

	private final ConcurrentMap<ServiceMetaData, PathChildrenCache> pathChildrenCaches = Maps.newConcurrentMap();
	// 指定节点都提供了哪些服务
	private final ConcurrentMap<ServiceAddress, ConcurrentSet<ServiceMetaData>> serviceMetaDataMap = Maps.newConcurrentMap();

	private CuratorFramework configClient;

	/**
	 * 链接注册中心
	 * 使用curator框架
	 * @param address 注册中心地址
	 */
	public void connect(String address) {
		configClient = CuratorFrameworkFactory.newClient(address, sessionTimeoutMs, connectionTimeoutMs, new ExponentialBackoffRetry(500, 20));
		configClient.getConnectionStateListenable().addListener((client, newState) -> {
			//如果是重连的情况则 重新订阅服务和发布服务
			if (newState == ConnectionState.RECONNECTED) {
				// 重新订阅
				for (ServiceMetaData serviceMetaData : getSubscribeSet()) {
					doSubscribe(serviceMetaData);
				}
				// 重新发布服务
				for (RegistryMetaData meta : getRegistryMetaDataMap().keySet()) {
					ZkRegistryServiceImpl.super.registry(meta);
				}
			}
		});
		configClient.start();
	}

	/**
	 * 注册服务
	 * @param meta
	 */
	protected void doRegistry(RegistryMetaData meta) {

		//服务在zk中的节点位置  eg: /ferry/provider/bj/userService/1.0
		String directory = String.format("/ferry/provider/%s/%s/%s",
				meta.getServiceMetaData().getGroup(),
				meta.getServiceMetaData().getServiceName(),
				meta.getServiceMetaData().getVersion());

		try {
			if (configClient.checkExists().forPath(directory) == null) {
				configClient.create().creatingParentsIfNeeded().forPath(directory);
			}
		} catch (Exception e) {
			logger.error("注册提供者持久节点异常, directory: {}", directory, e);
		}

		//注册临时节点
		try {
			meta.getServiceAddress().setIp(ipAddress);

			configClient.create().withMode(CreateMode.EPHEMERAL).inBackground((client, event) -> {
				//如果成功 则把注册节点状态改为done
				if (event.getResultCode() == KeeperException.Code.OK.intValue()) {
					getRegistryMetaDataMap().put(meta, RegistryState.DONE);
				}
				logger.info("注册临时节点成功, {}, {}", meta, event);
			}).forPath(
					String.format("%s/%s:%s",
							directory,
							meta.getServiceAddress().getIp(),
							String.valueOf(meta.getServiceAddress().getPort())
							));
		} catch (Exception e) {
			logger.warn("注册临时节点失败, meta:{}, e:{}", meta, e);
		}

	}

	/**
	 * 检查注册节点状态
	 */
	protected void doCheckNodeStatus() {

		for (Map.Entry<RegistryMetaData, RegistryState> entry : getRegistryMetaDataMap().entrySet()) {
			if (entry.getValue() == RegistryState.DONE) {
				continue;
			}

			//  /ferry/provider/bj/userService/1.0
			RegistryMetaData meta = entry.getKey();
			String directory = String.format("/ferry/provider/%s/%s/%s",
					meta.getServiceMetaData().getGroup(),
					meta.getServiceMetaData().getServiceName(),
					meta.getServiceMetaData().getVersion());

			String nodePath = String.format("%s/%s:%s",
					directory,
					meta.getServiceAddress().getIp(),
					String.valueOf(meta.getServiceAddress().getPort()));

			//重新注册临时节点
			try {
				if (configClient.checkExists().forPath(nodePath) == null) {
					super.registry(meta);
				}
			} catch (Exception e) {
				logger.warn("重新注册服务节点异常 meta:{}, e:{}", meta, e);
			}
		}

	}

	protected void doCancelRegistry(RegistryMetaData meta) {

		String directory = String.format("/ferry/provider/%s/%s/%s",
				meta.getServiceMetaData().getGroup(),
				meta.getServiceMetaData().getServiceName(),
				meta.getServiceMetaData().getVersion());

		try {
			if (configClient.checkExists().forPath(directory) == null) {
				return;
			}
		} catch (Exception e) {
			logger.warn("检查节点是否存在异常, directory: {}", directory, e);
		}

		try {
			meta.getServiceAddress().setIp(ipAddress);

			configClient.delete().inBackground((client, event) -> {
				logger.info("取消注册节点: {} {}.", meta, event);
			}).forPath(
					String.format("%s/%s:%s",
							directory,
							meta.getServiceAddress().getIp(),
							String.valueOf(meta.getServiceAddress().getPort())));
		} catch (Exception e) {
			logger.warn("删除注册节点异常: meta:{}, e:{}", meta, e);
		}

	}

	@Override
	public List<RegistryMetaData> query(ServiceMetaData serviceMeta) {
		String directory = String.format("/ferry/provider/%s/%s",
				serviceMeta.getServiceName(),
				serviceMeta.getVersion());

		List<RegistryMetaData> registerMetaList = Lists.newArrayList();
		try {
			List<String> paths = configClient.getChildren().forPath(directory);
			for (String p : paths) {
				registerMetaList.add(parseRegistryMeta(String.format("%s/%s", directory, p)));
			}
		} catch (Exception e) {
			if (logger.isWarnEnabled()) {
				logger.warn("Lookup service meta: {} path failed, {}.", serviceMeta, stackTrace(e));
			}
		}
		return registerMetaList;
	}

	/**
	 * 构建注册信息
	 * @param format
	 * @return
	 *
	 * /ferry/provider/bj/userService/1.0/192.168.0.1:9999
	 */
	private RegistryMetaData parseRegistryMeta(String format) {
		String[] array = format.split("/");
		ServiceMetaData serviceMetaData = new ServiceMetaData(array[3], array[4], array[5]);

		String[] array6 = array[6].split(":");
		ServiceAddress serviceAddress = new ServiceAddress(array6[0], Integer.valueOf(array6[1]));
		RegistryMetaData registryMetaData = new RegistryMetaData(serviceAddress, serviceMetaData);
		return registryMetaData;
	}

	/**
	 * 关闭注册器
	 */
	public void doShutDown() {
		for (PathChildrenCache childrenCache : pathChildrenCaches.values()) {
			try {
				childrenCache.close();
			} catch (IOException ignored) {}
		}

		configClient.close();
	}

	/**
	 * 订阅服务 消费者使用
	 * @param serviceMetaData
	 */
	public void doSubscribe(ServiceMetaData serviceMetaData) {

		PathChildrenCache childrenCache = pathChildrenCaches.get(serviceMetaData);
		if (childrenCache == null) {
			String directory = String.format("/ferry/provider/%s/%s/%s",
					serviceMetaData.getGroup(),
					serviceMetaData.getServiceName(),
					serviceMetaData.getVersion());

			PathChildrenCache newChildrenCache = new PathChildrenCache(configClient, directory, false);
			childrenCache = pathChildrenCaches.putIfAbsent(serviceMetaData, newChildrenCache);
			if (childrenCache == null) {
				childrenCache = newChildrenCache;

				childrenCache.getListenable().addListener((client, event) -> {

					switch (event.getType()) {
						//服务节点上线
						case CHILD_ADDED: {
							//新注册节点
							RegistryMetaData registryMetaData = parseRegistryMeta(event.getData().getPath());
							ServiceAddress serviceAddress = registryMetaData.getServiceAddress();
							ServiceMetaData registryMetaDataServiceMeta = registryMetaData.getServiceMetaData();
							ConcurrentSet<ServiceMetaData> serviceMetaDataSet = getServiceMetaData(serviceAddress);
							serviceMetaDataSet.add(registryMetaDataServiceMeta);

							//通知监听者
							ZkRegistryServiceImpl.super.notify(
									registryMetaDataServiceMeta,
									NodeEvent.LEAF_ADD,
									Arrays.asList(registryMetaData));

							break;
						}
						//服务节点下线
						case CHILD_REMOVED: {
							RegistryMetaData registryMetaData = parseRegistryMeta(event.getData().getPath());
							ServiceAddress serviceAddress = registryMetaData.getServiceAddress();
							ServiceMetaData registryMetaDataServiceMeta = registryMetaData.getServiceMetaData();
							ConcurrentSet<ServiceMetaData> serviceMetaSet = getServiceMetaData(serviceAddress);
							serviceMetaSet.remove(registryMetaDataServiceMeta);

							ZkRegistryServiceImpl.super.notify(
									registryMetaDataServiceMeta,
									NodeEvent.LEAF_REMOVED,
									Arrays.asList(registryMetaData));

							if (serviceMetaSet.isEmpty()) {
								ZkRegistryServiceImpl.super.offline(serviceAddress);
							}
							break;
						}
					}
				});

				try {
					childrenCache.start();
				} catch (Exception e) {
					if (logger.isWarnEnabled()) {
						logger.warn("Subscribe {} failed, {}.", directory, stackTrace(e));
					}
				}
			} else {
				try {
					newChildrenCache.close();
				} catch (IOException e) {
					logger.warn("关闭新的监听器异常, directory:{}, e:{}", directory, e);
				}
			}
		}

	}

	/**
	 * 根据地址获取服务列表
	 * @param serviceAddress
	 * @return
	 */
	private ConcurrentSet<ServiceMetaData> getServiceMetaData(ServiceAddress serviceAddress) {
		ConcurrentSet<ServiceMetaData> serviceMetaSet = serviceMetaDataMap.get(serviceAddress);
		if (serviceMetaSet == null) {
			ConcurrentSet<ServiceMetaData> newServiceMetaSet = new ConcurrentSet<>();
			serviceMetaSet = serviceMetaDataMap.putIfAbsent(serviceAddress, newServiceMetaSet);
			if (serviceMetaSet == null) {
				serviceMetaSet = newServiceMetaSet;
			}
		}
		return serviceMetaSet;
	}

}
